package com.shujia.flink.core;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo9Submit {
    public static void main(String[] args) throws Exception {
        //1、创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2、读取数据
        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        //使用lambda表达式处理数据
        DataStream<String> wordsDS = linesDS
                .flatMap((line, out) -> {
                    for (String word : line.split(",")) {
                        out.collect(word);
                    }
                }, Types.STRING);

        DataStream<Tuple2<String, Integer>> kvDS = wordsDS
                .map(word -> Tuple2.of(word, 1))
                //指定返回类型
                .returns(Types.TUPLE(Types.STRING, Types.INT));

        KeyedStream<Tuple2<String, Integer>, String> keyByDS = kvDS.keyBy(kv -> kv.f0);

        //kv1代表之前的结果（状态），kv2代码最新一条数据
        //reduce：有状态计算
        DataStream<Tuple2<String, Integer>> countDS = keyByDS
                .reduce((kv1, kv2) -> {
                    return Tuple2.of(kv1.f0, kv1.f1 + kv2.f1);
                });

        countDS.print();

        //execute方法会触发任务执行（任务调度）
        env.execute("lambda");
    }
}
